package com.calvin.study.service.impl;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import org.redisson.api.RLock;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.cloud.commons.lang.StringUtils;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.calvin.study.feign.ProductFeignClient;
import com.calvin.study.mapper.SeckillActMapper;
import com.calvin.study.mapper.SeckillProductMapper;
import com.calvin.study.rabbitmq.producer.RabbitMqProducer;
import com.calvin.study.service.ISeckillService;

import cloud.alibaba.study.constants.RedisConstant;
import cloud.alibaba.study.entity.base.ResponseVo;
import cloud.alibaba.study.entity.order.Order;
import cloud.alibaba.study.entity.product.Product;
import cloud.alibaba.study.entity.seckill.SeckillAct;
import cloud.alibaba.study.entity.seckill.SeckillProduct;
import cloud.alibaba.study.utils.DateUtil;
import cloud.alibaba.study.utils.GsonUtil;
import cloud.alibaba.study.vo.seckill.SeckillProductVo;

@Service
public class SeckillService implements ISeckillService {

	@Autowired
	private SeckillActMapper seckillActMapper;
	@Autowired
	private SeckillProductMapper seckillProductMapper;
	@Autowired
	StringRedisTemplate redisTemplate;
	@Autowired
	ProductFeignClient productFeignClient;
	@Autowired
	RedissonClient redissonClient;
	@Autowired
	RabbitMqProducer rabbitMqProducer;

	@Transactional
	public void seckillProductShelves(int day) {
		List<SeckillAct> seckillActs = getDaysLaterSeckillAct(day);

		if (!CollectionUtils.isEmpty(seckillActs)) {
			try {
				//这块redis应该用事务来控制
				seckillActToRedis(seckillActs);
				
				//修改秒杀活动状态
				List<Long> seckillActIds = seckillActs.stream().map(item -> item.getSeckillActId()).collect(Collectors.toList());
				seckillActMapper.batchUpdateStatus(seckillActIds);
				
				Map<Long,Integer> map = new HashMap<Long,Integer>();
				//商品上架后，远程通知库存服务，将秒杀活动库存预占。
				seckillActs.stream().forEach(act -> {
					
					act.getSeckillProducts().stream().forEach(seckillproduct -> {
						Long productId = seckillproduct.getProductId();
						if(map.containsKey(productId)) {
							Integer count = map.get(productId);
							map.put(productId, count + seckillproduct.getSeckillCount());
						}
						map.put(productId, seckillproduct.getSeckillCount());
					});
					
					Long dateTime = act.getEndTime().getTime() - act.getStartTime().getTime();
					//将所有秒杀的活动推送到MQ中，利用结束时间作为消息的过期时间，活动结束后，对库存进行结算
					rabbitMqProducer.pushToExchange("seckill-event-exchange", "seckill.act.and", GsonUtil.toJsonDateFormat(act), "seckill_act_" + act.getSeckillActId(), String.valueOf(dateTime));
				});
				
				//针对每个秒杀活动的商品的库存计算，通知库存服务进行预占
				map.forEach((k,v)->{
					Order order = new Order();
					order.setProductId(k);
					order.setCount(v);
					rabbitMqProducer.pushToExchange("storage-event-exchange", "storage.reserved", order,String.valueOf(order.getOrderId()));
				});
				
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	// 查询从当天开始，近X天的秒杀活动（以及对应的秒杀商品信息）
	public List<SeckillAct> getDaysLaterSeckillAct(int day) {
		List<SeckillAct> seckillActs = null;

		QueryWrapper<SeckillAct> qw = new QueryWrapper<SeckillAct>();
		qw.between("start_time", DateUtil.getStartDateTime(), DateUtil.getDaysLaterDateTime(day)).eq("status", 0);
		List<SeckillAct> actList = seckillActMapper.selectList(qw);
		
		if (!CollectionUtils.isEmpty(actList)) {
			seckillActs = actList.stream().map(seckillproduct -> {
				// 查询秒杀活动关联的商品
				QueryWrapper<SeckillProduct> q = new QueryWrapper<SeckillProduct>();
				q.eq("seckill_act_id", seckillproduct.getSeckillActId());
				List<SeckillProduct> seckillProduct = seckillProductMapper.selectList(q);
				seckillproduct.setSeckillProducts(seckillProduct);
				return seckillproduct;
			}).collect(Collectors.toList());
		}

		return seckillActs;
	}

    /**
     * 该方法缓存商品信息
     * 有两个重点：随机码、信号量（库存数）
     * 随机码是用于商品加密，只有在秒杀时间段才会返回给前台，该值作为库存的key
     * 使用redisson将库存数作为信号量
     *
     * 三个redis数据结构
     * list数据格式[活动编号]_[开始时间]_[结束时间] 
     * list key:SECKILL:ACT:4_1658620800000_1658628000000 val:1(商品id)
     * hash key:SECKILL:PRODUCT:  key(1-1) value(SeckillProduct)
     * string SECKILL:PRODUCT:STOCK(key): (value)stockNum
     */
	private void seckillActToRedis(List<SeckillAct> seckillActs) {
		if (!CollectionUtils.isEmpty(seckillActs)) {
			seckillActs.stream().forEach(seckillact -> {
				// 生成秒杀活动信息缓存，采用redis list数据结构存储
				Long startTime = seckillact.getStartTime().getTime();
				Long endTime = seckillact.getEndTime().getTime();
				//避免时间相同，活动不同，只生成一场活动信息 
				String key = RedisConstant.SECKILL_ACT_CACHE_PREFIX + seckillact.getSeckillActId()+ "_" + startTime + "_" + endTime;

				if (Boolean.FALSE.equals(redisTemplate.hasKey(key))) {
					// 没有上架的商品才会到这，【所以同一个商品只能加入到一个秒杀场次中】
					List<String> productIds = seckillact.getSeckillProducts().stream()
							.map(item -> item.getProductId().toString())
							.collect(Collectors.toList());

					redisTemplate.opsForList().leftPushAll(key, productIds);
				}
				
				//生成秒杀商品的key
				BoundHashOperations<String, String, String> ops = redisTemplate.boundHashOps(RedisConstant.SECKILL_PRODUCT_CACHE_PREFIX);
				
				// 遍历秒杀商品信息缓存，采用redis hash数据结构存储
				seckillact.getSeckillProducts().stream().forEach(seckillproduct -> {
					SeckillProductVo spv = null;
					String productKey = seckillproduct.getSeckillActId() + "_" + seckillproduct.getProductId().toString();
					if (Boolean.FALSE.equals(ops.hasKey(productKey))) {
						spv = new SeckillProductVo();
						// 秒杀信息
						BeanUtils.copyProperties(seckillproduct, spv);

						//远程调用查询商品基础信息
						ResponseVo resp = productFeignClient.getDetail(seckillproduct.getProductId());
						if (!resp.getIsOk()) {
							throw new RuntimeException(resp.getRespMsg());
						}
						Product product = resp.getRespDate(resp.getRespData(), Product.class);
						spv.setProductInfo(product);
						//设置当前商品的秒杀时间信息
						spv.setStartTime(seckillact.getStartTime().getTime());
						spv.setEndTime(seckillact.getEndTime().getTime());
						 //设置商品随机码
						spv.setRandomCode(UUID.randomUUID().toString().replace("-", ""));//防止恶意行为刷单，保护机制
						//商品信息设置到缓存中
						ops.put(productKey,GsonUtil.toJsonDateFormat(spv));

						// 如果当前这个场次商品的库存信息已经上架就无需上架
						// 使用库存作为分布式信号量 ，限流，避免直接查询数据库
						// 为什么key使用随机码生成，处于保护机制，只有在用户下单后，获取随机码时，开始减扣库存，避免恶意行为刷单
						RSemaphore semaphore = redissonClient.getSemaphore(RedisConstant.SECKILL_PRODUCT_STOCK_SEMAPHORE + spv.getRandomCode());
						semaphore.trySetPermits(seckillproduct.getSeckillCount());// 秒杀商品数量
					}
				});
				
			});
		}
	}

	private void seckillProductToRedis(List<SeckillAct> seckillActs) throws Exception {
		if (!CollectionUtils.isEmpty(seckillActs)) {
			// 遍历秒杀活动信息
			seckillActs.stream().forEach(seckillact -> {
				BoundHashOperations<String, Object, Object> ops = redisTemplate .boundHashOps(RedisConstant.SECKILL_PRODUCT_CACHE_PREFIX);

				seckillact.getSeckillProducts().stream().forEach(seckillproduct -> {// 遍历秒杀商品信息
					SeckillProductVo spv = null;

					if (!ops.hasKey(
							seckillproduct.getSeckillActId() + "_" + seckillproduct.getProductId().toString())) {
						spv = new SeckillProductVo();
						// 秒杀信息
						BeanUtils.copyProperties(seckillproduct, spv);

						// 查询商品基础信息
						ResponseVo resp = productFeignClient.getDetail(seckillproduct.getProductId());
						if (!resp.getIsOk()) {
							throw new RuntimeException(resp.getRespMsg());
						}
						Product product = resp.getRespDate(resp.getRespData(), Product.class);

						spv.setProductInfo(product);

						spv.setStartTime(seckillact.getStartTime().getTime());
						spv.setEndTime(seckillact.getEndTime().getTime());
						spv.setRandomCode(UUID.randomUUID().toString().replace("-", ""));//防止恶意行为刷单，保护机制

						ops.put(seckillproduct.getSeckillActId() + "_" + seckillproduct.getProductId().toString(),
								GsonUtil.toJsonDateFormat(spv));

						// 如果当前这个场次商品的库存信息已经上架就无需上架
						// 使用库存作为分布式信号量 ，限流，避免直接查询数据库
						// 为什么key使用随机码生成，处于保护机制，只有在用户下单后，获取随机码时，开始减扣库存，避免恶意行为刷单
						RSemaphore semaphore = redissonClient.getSemaphore(RedisConstant.SECKILL_PRODUCT_STOCK_SEMAPHORE + spv.getRandomCode());
						semaphore.trySetPermits(seckillproduct.getSeckillCount());// 秒杀商品数量
					}
				});
			});
		}
	}
	
	//https://gitee.com/JuneQ/mall-project/blob/master/mall-seckill/src/main/java/org/june/seckill/service/impl/SeckillServiceImpl.java
	//https://gitee.com/AdverseQ/gulimall_Advanced
	
	
	/**
	 * 获取当前时间所有的可秒杀商品
	 */
	public List<List<SeckillProductVo>> getCurrentSeckillProducts() {
		List<List<SeckillProductVo>> seckillProducts = new ArrayList<List<SeckillProductVo>>();
		//获取所有的秒杀活动
		//1) "SECKILL:ACT:1_1659542400000_1659887999000"
		//2) "SECKILL:ACT:2_1659542400000_1659801599000"                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 
		Set<String> keys = redisTemplate.keys(RedisConstant.SECKILL_ACT_CACHE_PREFIX + "*");
		if (!CollectionUtils.isEmpty(keys)) {
			for (String key : keys) {                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  
				long now = DateUtil.getCurDate2Long();
				//截取活动key的参数
				String[] splitParam = key.replace(RedisConstant.SECKILL_ACT_CACHE_PREFIX, "").split("_");
				String act = splitParam[0];//活动ID
				long start = Long.parseLong(splitParam[1]);//秒杀开始时间
				long end = Long.parseLong(splitParam[2]);//秒杀结束时间
				// 符合当前时间场次的秒杀活动
				if (now >= start && now <= end) {
					//获取秒杀活的Val（商品的ID：1,2）
					List<String> productIds = redisTemplate.opsForList().range(key, -100, 100);
					
					if (!CollectionUtils.isEmpty(productIds)) {
						//从缓存中获取秒杀商品详细信息
						BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(RedisConstant.SECKILL_PRODUCT_CACHE_PREFIX);
                        //生成hash结构的 keys 由（活动id_商品id)拼接
                        List<String> hashKeys = productIds.stream().map(p -> p= act +"_" +p).collect(Collectors.toList());
                        // hash结构的value //{"seckillProductId":2,"seckillActId":2,"productId":2,"seckillPrice":599.00,"seckillCount":15,"seckillLimit":1,"seckillSort":null,"createTime":"2022-08-01T16:00:00.000+00:00","productInfo":{"productId":2,"productName":"华硕圣骑士 GTX750TI 2G显存 双风扇散热","storageId":2,"productPrice":650.0,"productStatus":1,"userId":1},"startTime":1659542400000,"endTime":1659801599000,"randomCode":null}
						List<String> projectJson = hashOps.multiGet(hashKeys);
						System.out.println(projectJson);
						if (!CollectionUtils.isEmpty(projectJson)) {
							List<SeckillProductVo> collect = projectJson.stream().map(project -> {
								//数据转换
								SeckillProductVo redis = GsonUtil.fromJson(project, SeckillProductVo.class);
								redis.setRandomCode(null);
								return redis;
							}).collect(Collectors.toList());
							
							seckillProducts.add(collect);
						}
					}
				}
			}
		}
		return seckillProducts;
	}
	
	
	/**
	 * 秒杀商品详情页
	 * 1.从缓存中获取所有秒杀商品的key进行遍历
	 * 2.将传入的两个参数拼接，与缓存中商品的hashkey匹配
	 * 3.按照当前时间与秒杀商品时间比对，是否返回秒杀库存量
	 */
	public SeckillProductVo getSeckillProductDetail(Long seckillActId,Long productId){
		SeckillProductVo seckillProduct = null;
		
		//等同于命令：HKEYS SECKILL:PRODUCT
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(RedisConstant.SECKILL_PRODUCT_CACHE_PREFIX);
        Set<String> hashKeys = hashOps.keys();
        if(CollectionUtils.isNotEmpty(hashKeys)) {
        	String key = seckillActId+"_"+productId; 
        	for(String hashKey : hashKeys) {
        		//参数拼接的Key和缓存中的key匹配
        		if(hashKey.equals(key)) {
        			String json = hashOps.get(key);
        			seckillProduct = GsonUtil.fromJson(json, SeckillProductVo.class, GsonUtil.DATE_FORMAT);
        			seckillProduct.setSkillId(key);
        			Long curTime = DateUtil.getCurDate2Long();
        			if(!(curTime>=seckillProduct.getStartTime() && curTime<=seckillProduct.getEndTime())) {
        				//秒杀未开始，库存信号量不能返回前台。
        				seckillProduct.setRandomCode(null);
        			}
        			return seckillProduct;
        		}
        	}
        }
		return seckillProduct;
	}
	
	public static void main(String[] args) {
		String regx = "2" + "_\\d";
		System.out.println(Pattern.matches(regx, "2_7"));
		
		System.out.println(12 << 1);
		System.out.println(1 << 30 );
		
		Map<String,String> map = new HashMap<>();
		map.put("1","5");
		map.put("2","10");
		map.put("3","8");
		map.forEach((k,v)->{
			System.out.println("k:" + k + " v:" + v);
		});
	}

	@Override
	public String doSeckill(String killId, String randomCode, Integer num) throws Exception {
		  BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(RedisConstant.SECKILL_PRODUCT_CACHE_PREFIX);
		  String json = hashOps.get(killId);
		  if(StringUtils.isBlank(json)) {
			  throw new Exception("无效的秒杀商品"); 
		  }
		  SeckillProductVo sp = GsonUtil.fromJson(json, SeckillProductVo.class, GsonUtil.DATE_FORMAT);
          Long startTime = sp.getStartTime();
          Long endTime = sp.getEndTime();
          Long curTime = DateUtil.getCurDate2Long();

		  if(null!=sp) {
			 //1.校验秒杀商品时间
			  if(!(curTime>=startTime && curTime <= endTime)) {
				  throw new Exception("秒杀失败，暂时无法秒杀该商品");
			  }
			//2.校验商品库存随机码
			  if(!randomCode.equals(sp.getRandomCode())) {
				  throw new Exception("秒杀失败，无效商品");
			  }
			//3.校验购买数量
			  if(num > sp.getSeckillLimit()) {
				  throw new Exception("秒杀失败，超量秒杀");
			  }
			//4.校验是否已完成秒杀
			  Long userId = 101869l;
			  //分布式锁key ： 101869_1_3_afdljalfjlajf
			  String seckillUserKey = RedisConstant.SECIKLL_USER_MARK_PREFIX +"_"+ userId + "_" + sp.getSeckillActId()+"_"+sp.getProductId() + "_" + sp.getRandomCode();
			  
			  
			  //判断用户是否秒杀过商品 ttl：Duration.ofMillis(endTime-curTime) 秒杀商品结束时间 - 当前参与秒杀的时间内 无法再次参与秒杀
			  //RLock rlock =redissonClient.getLock(seckillUserKey);
			  //rlock.lock(endTime-curTime, TimeUnit.MILLISECONDS);
			  try {
				  Boolean hasSeckill = redisTemplate.opsForValue().setIfAbsent(seckillUserKey, String.valueOf(num),Duration.ofMillis(endTime-curTime));
				  if(Boolean.FALSE.equals(hasSeckill)) {
					  throw new Exception("已秒杀过该商品，无法再次参与~");
				  }
			  
				  RSemaphore stockSemaphore = redissonClient.getSemaphore(RedisConstant.SECKILL_PRODUCT_STOCK_SEMAPHORE + randomCode);
			  
                  if (stockSemaphore.tryAcquire(num, 500, TimeUnit.MILLISECONDS)) {
                      // 秒杀成功
                      String orderNo = IdWorker.getTimeId();
                      
                      // 对象构造
                      Order seckillOrder = new Order();
                      seckillOrder.setOrderNo(orderNo);
                      seckillOrder.setProductId(sp.getProductId());
                      seckillOrder.setUnitPrice(sp.getSeckillPrice());
                      seckillOrder.setCount(num);
                      seckillOrder.setTotalPrice(sp.getSeckillPrice().multiply(new BigDecimal(num)));
                      seckillOrder.setUserId(userId);
                      seckillOrder.setStatus(0);
                      seckillOrder.setOrderType(20);//秒杀类型订单
                      // 秒杀成功，发送mq订单消息，并返回订单号通知前台秒杀成功
                      rabbitMqProducer.pushToExchange("order-event-exchange", "order.seckill.order", seckillOrder, orderNo);
                      return orderNo;
                  }else {
                	  throw new Exception("商品已售罄~");
                  }
              } catch (InterruptedException e) {
            	  e.printStackTrace();
            	  throw new Exception("秒杀失败");
              }finally{
            	// 还在持有锁的状态，并且是当前线程持有的锁再解锁
      			//if (rlock.isLocked() && rlock.isHeldByCurrentThread()) {
      				//rlock.unlock();
      			//}
              }
		  }
		return "";
	}


}
